package cc.yugu.yhdw.webserver.comm.client;

import cc.yugu.yhdw.common.constant.ConstEnum;
import cc.yugu.yhdw.common.constant.Constant;
import cc.yugu.yhdw.common.tool.netty.Message;
import cc.yugu.yhdw.common.tool.zk.Listener;
import cc.yugu.yhdw.common.utils.YhdwUtils;
import cc.yugu.yhdw.webserver.config.AppConfig;
import io.netty.channel.ChannelFuture;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;

import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
/**
 * service-server的客户端管理
 * 
 * @Date:Created in 2018/6/26 14:05
 * @Author:
 * @Modified By: zjn 2018/6/26 14:05
 */
public class ServiceServerClientManager implements Listener {
    static Logger log = LogManager.getLogger(ServiceServerClientManager.class);

    final Map<String, ServiceServerClient> clientMap = new ConcurrentHashMap<String, ServiceServerClient>();
    AppConfig config;

    public ServiceServerClientManager(AppConfig config) {
        this.config = config;
    }
    /**
    * 写入和刷新
    * ChannelFuture 类：与未来相关联的I/O操作的通道
    * @Param: [address, message]
    * @Return: io.netty.channel.ChannelFuture
    * @Author By: zjn 2018/6/27  10:03
    */
    public ChannelFuture writeAndFlush(String address, Message message) {
        ServiceServerClient terminalServerClient = clientMap.get(address);
        if(terminalServerClient == null || terminalServerClient.getChannel() == null || !terminalServerClient.getChannel().isOpen()) {
            return null;
        }

        return terminalServerClient.getChannel().writeAndFlush(message);
    }

    public ChannelFuture write(String address, Message message) {
        ServiceServerClient terminalServerClient = clientMap.get(address);
        if(terminalServerClient == null || terminalServerClient.getChannel() == null || !terminalServerClient.getChannel().isOpen()) {
            return null;
        }

        return terminalServerClient.getChannel().write(message);
    }

    @Override
    public void listen(Param param) throws Exception {
        List<String> children = (List<String>) param.result;
        Set<String> servers = new HashSet<String>(10);

        for(String child : children) {
            byte[] data = config.zkClient.get(ConstEnum.Node.NODE_SERVICE_SERVER.getValue() + "/" + child);
            String source = new String(data, Constant.CHARSET_UTF_8);
            Map<String, String> map = YhdwUtils.stringToMap(source, ",");

            if(log.isDebugEnabled()) {
                log.debug("child: {}", source);
            }

            String host = map.get("server.ip");
            int port = Integer.parseInt(map.get("server.port"));
            String key = host + ":" + port;
            String threadName = "ServiceServerClient/" + key;
            servers.add(key);

            if(!clientMap.containsKey(key)) {
                ServiceServerClient client = new ServiceServerClient(host, port, threadName, config.bizFactory, config.executorService);
                client.startup();

                clientMap.put(key, client);
            }
        }

        for(String key : clientMap.keySet()) {
            if(!servers.contains(key)) {
                ServiceServerClient client = clientMap.remove(key);
                if(client != null) {
                    client.close();
                }
            }
        }
    }
}
